package com.codeduck.kafka.kafka.producer;

import com.codeduck.kafka.exception.MessageException;
import com.codeduck.kafka.kafka.properties.ProducerProperties;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Properties;

/**
 * @author: <a href="mailto:codeduck@163.com">Codeduck</a>
 * @date: 2023/3/1
 */
@Component
@Scope(BeanDefinition.SCOPE_PROTOTYPE)
public class KafkaMessageProducer implements IMessageProducer {

    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaMessageProducer.class);
    private ProducerRecord record;

    @Override
    public void sendMessage(String message, String topic) throws MessageException {
        this.sendMessage(message, topic, -1);
    }

    @Override
    public void sendMessage(String message, String topic, int partition) throws MessageException {
        this.send(message, topic, partition);
    }

    private void send(String message, String topic, int partition) throws MessageException {
        if (partition > -1) {
            record = new ProducerRecord(topic, partition, null, message);
        } else {
            record = new ProducerRecord(topic, message);
        }
        this.doSend(ProducerProperties.getProperties());
    }

    private void doSend(Properties properties) throws MessageException {
        KafkaProducer<Object, Object> producer = new KafkaProducer<>(properties);
        try {
            // 异步发送消息
            producer.send(this.record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (null != exception) {
                        LOGGER.error("Kafka 消息发送至 Topic[{}]失败", metadata.topic(), exception);
                    } else {
                        LOGGER.info("============> Kafka 成功发送消息至 Topic[{}] <============", metadata.topic());
                    }
                }
            });
        } catch (Exception e) {
            LOGGER.error("Kafka 消息发送过程中出现异常", e);
            throw new MessageException(e);
        } finally {
            producer.close();
        }
    }
}
